package com.nbcio.iot.listener;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.nbcio.iot.queue.IotMessageQueueService;
import com.nbcio.iot.service.IotRuleExecutionService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * IoT 消息队列监听器
 * 监听数据采集消息并转发到规则执行队列
 *
 * @author nbacheng
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class IotMessageQueueListener {

    private final IotMessageQueueService messageQueueService;
    private final IotRuleExecutionService ruleExecutionService;
    private final ObjectMapper objectMapper = new ObjectMapper();
    
    private final AtomicBoolean running = new AtomicBoolean(false);
    private volatile Thread listenerThread;

    @PostConstruct
    public void startListening() {
        if (running.compareAndSet(false, true)) {
            listenerThread = new Thread(this::listenDataCollectionMessages, "iot-queue-listener");
            listenerThread.setDaemon(true);
            listenerThread.start();
            log.info("[QUEUE-LISTENER] 消息队列监听器已启动");
        }
    }

    @PreDestroy
    public void stopListening() {
        if (running.compareAndSet(true, false)) {
            log.info("[QUEUE-LISTENER] 正在停止消息队列监听器...");
            if (listenerThread != null) {
                listenerThread.interrupt();
                try {
                    listenerThread.join(5000); // 等待最多5秒
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
            log.info("[QUEUE-LISTENER] 消息队列监听器已停止");
        }
    }

    /**
     * 监听数据采集消息
     */
    private void listenDataCollectionMessages() {
        while (running.get() && !Thread.currentThread().isInterrupted()) {
            try {
                String messageJson = messageQueueService.consumeDataCollectionMessage();
                if (messageJson != null) {
                    Map<String, Object> message = objectMapper.readValue(messageJson, Map.class);
                    
                    // 将数据采集消息转换为规则执行消息
                    message.put("messageType", "RULE_EXECUTION");
                    
                    // 发送到规则执行队列
                    messageQueueService.sendRuleExecutionMessage(message);
                    
                    log.info("[QUEUE-LISTENER] 数据采集消息已转发到规则执行队列");
                }
            } catch (Exception e) {
                if (running.get()) {
                    log.error("[QUEUE-LISTENER] 处理数据采集消息失败: {}", e.getMessage());
                    try {
                        Thread.sleep(1000); // 出错时等待1秒
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                        log.info("[QUEUE-LISTENER] 等待线程被中断，正在退出...");
                        break;
                    }
                }
            }
        }
        log.info("[QUEUE-LISTENER] 监听线程已退出");
    }
}
